﻿//TransactionManager.cs
//Copyright (c) 2013 StockSharp LLC, all rights reserved.
//This code module is part of StockSharp library.
//This code is licensed under the GNU GENERAL PUBLIC LICENSE Version 3.
//See the file License.txt for the license details.
//More info on: http://stocksharp.com

namespace StockSharp.Plaza
{
	using System;
	using System.Collections.Generic;
	using System.Linq;

	using Ecng.Collections;
	using Ecng.Common;

	using StockSharp.Algo;
	using StockSharp.Plaza.Wrappers;

	/// <summary>
	/// Менеджер транзакций.
	/// </summary>
	public class TransactionManager
	{
		private readonly MessageDispatcher _messageDispatcher;
		private readonly PlazaConnectionPool _connectionPool;

		private readonly SynchronizedMultiDictionary<long, Transaction> _transactionsByTransactionIds = new SynchronizedMultiDictionary<long, Transaction>();

		private readonly SynchronizedQueue<Transaction> _transactionQueue = new SynchronizedQueue<Transaction>();
		private readonly SynchronizedList<PlazaConnection> _connections = new SynchronizedList<PlazaConnection>();

		internal TransactionManager(PlazaConnectionPool connectionPool, TransactionIdGenerator idGenerator)
		{
			if (connectionPool == null)
				throw new ArgumentNullException("connectionPool");

			_connectionPool = connectionPool;

			Factory = new TransactionFactory(connectionPool, idGenerator);
			TransactionTimeOut = TimeSpan.FromSeconds(5);

			_messageDispatcher = new MessageDispatcher((msg, errorCode, transactionId) =>
			{
				var transaction = GetTransactions(transactionId).Single();

				if (errorCode != 0)
				{
					//http://forum.rts.ru/viewtopic.asp?p=118911
					TransactionFailed.SafeInvoke(transaction, errorCode);
				}
				else
				{
					ProcessReply(transaction, msg);
				}
			});

			IsAsync = true;

			// Необходимо создать хотя бы одно транзакционное подключение
			ThreadCount = 1;
		}

		/// <summary>
		/// Использовать ли асинхронный режим при отправке транзакций.
		/// </summary>
		/// <remarks>
		/// Значение по умолчанию равно true.
		/// </remarks>
		public bool IsAsync { get; set; }

		/// <summary>
		/// Фабрика транзакций.
		/// </summary>
		public TransactionFactory Factory { get; private set; }

		/// <summary>
		/// Время, в течении которого ожидается ответ для транзакции.
		/// </summary>
		/// <remarks>
		/// По-умолчанию равно 5 секундам.
		/// </remarks>
		public TimeSpan TransactionTimeOut { get; set; }

		private int _threadCount;

		/// <summary>
		/// Число транзакционных потоков.
		/// </summary>
		/// <remarks>
		/// Значение по умолчанию равно 1.
		/// </remarks>
		public int ThreadCount
		{
			get { return _threadCount; }
			set
			{
				if (value < 1)
					throw new ArgumentOutOfRangeException("value", value, @"Число транзакционных потоков не может быть меньше 1.");

				if (_connections.Count > 0)
					throw new InvalidOperationException("Нельзя менять количество транзакционных потоков после из запуска.");

				if (value == ThreadCount)
					return;

				_threadCount = value;
			}
		}

		/// <summary>
		/// Обработать отправляемую транзакцию до того, как она будет отослана на сервер Plaza.
		/// </summary>
		public event Action<Transaction> ProcessRequest;

		/// <summary>
		/// Обработать полученную транзакцию до того, как она будет обработана <see cref="PlazaTrader"/>.
		/// </summary>
		public event Action<Transaction> ProcessResponse;

		/// <summary>
		/// Ответ на посланную транзакцию не успел придти за <see cref="TransactionTimeOut"/>.
		/// </summary>
		public event Action<Transaction, uint> TransactionFailed;

		private bool _useTransactQueue = true;

		/// <summary>
		/// Если true, транзакция буферизируется во внутренюю очередь, обрабатываемая с частотой <see cref="PlazaTrader.PollTimeOut"/>.
		/// Если false, транзакция напрямую отправляется в роутер.
		/// По умолчанию true.
		/// </summary>
		public bool UseTransactionsQueue
		{
			get { return _useTransactQueue; }
			set { _useTransactQueue = value; }
		}

		/// <summary>
		/// Отправить транзакцию на сервер Plaza.
		/// </summary>
		/// <param name="transaction">Транзакция.</param>
		public void SendTransaction(Transaction transaction)
		{
			if (transaction == null)
				throw new ArgumentNullException("transaction");

			ProcessRequest.SafeInvoke(transaction);

			AddTransaction(transaction);

			if (UseTransactionsQueue)
				_transactionQueue.Enqueue(transaction);
			else
			{
				var connection = _connections.SyncGet(c => c.FirstOrDefault());

				if (connection == null)
					throw new InvalidOperationException("Не создано ни одного подключения.");

				SendTransaction(connection, transaction);
			}
		}

		private void SendTransaction(PlazaConnection connection, Transaction transaction)
		{
			if (connection == null)
				throw new ArgumentNullException("connection");

			if (transaction == null)
				throw new ArgumentNullException("transaction");

			var timeOut = (uint)TransactionTimeOut.TotalMilliseconds;
			var plazaMessage = transaction.PlazaMessage;

			if (IsAsync)
			{
				plazaMessage.SendAsync(connection.NativeConnection, timeOut, _messageDispatcher, transaction.Id);
			}
			else
			{
				var reply = plazaMessage.Send(connection.NativeConnection, timeOut);
				ProcessReply(transaction, reply);
			}
		}

		internal void Start()
		{
			lock (_connections.SyncRoot)
			{
				if (_connections.Count > 0)
				{
					if (_connections.Any(c => c.NativeConnection.Status != ConnectionWrapperStatus.Disconnected))
						throw new InvalidOperationException("Транзакционные потоки не были остановлены с предыдущего запуска.");
					else
						_connections.Clear();
				}

				var connections = new List<PlazaConnection>();

				for (var i = 0; i < ThreadCount; i++)
				{
					connections.Add(_connectionPool.RunConnection("Plaza. Transaction thread #{0}".Put(i + 1), connection => { }, (connection, processError) =>
					{
						while (true)
						{
							var newTransaction = _transactionQueue.TryDequeue();
							if (newTransaction == null)
								break;

							SendTransaction(connection, newTransaction);
						}
					}, connection => { }));
				}

				foreach (var connection in connections)
				{
					connection.Start();
					_connections.Add(connection);
				}
			}
		}

		internal void Stop()
		{
			lock (_connections.SyncRoot)
			{
				_connections.ForEach(c => c.Stop(() => _connections.Remove(c)));
			}
		}

		/// <summary>
		/// Получить транзакции (запрос и ответ) по номеру.
		/// </summary>
		/// <param name="transactionId">Номер транзакции.</param>
		/// <returns>Транзакции (запрос и ответ).</returns>
		public IEnumerable<Transaction> GetTransactions(long transactionId)
		{
			return _transactionsByTransactionIds.TryGetValue(transactionId);
		}

		private void AddTransaction(Transaction transaction)
		{
			if (transaction == null)
				throw new ArgumentNullException("transaction");

			_transactionsByTransactionIds.Add(transaction.Id, transaction);
		}

		private void ProcessReply(Transaction originMessage, IMessageWrapper reply)
		{
			var replyTransaction = new Transaction(originMessage, reply);
			AddTransaction(replyTransaction);

			PlazaException errorInfo;

			if (replyTransaction.GetCategory() == TransactionFactory.MessagesCategory)
			{
				var code = replyTransaction.GetReplyCode();
				errorInfo = code == 0 ? null : new PlazaException(code, replyTransaction.GetMessage());
			}
			else
				errorInfo = new PlazaException(replyTransaction.GetErrorCode(), "Неизвестная ошибка.");

			replyTransaction.ErrorInfo = errorInfo;

			ProcessResponse.SafeInvoke(replyTransaction);
		}
	}
}